package com.shujia.hbase.api;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryPrefixComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;

public class Demo12Index {

    private HConnection connection;
    private HBaseAdmin hBaseAdmin;

    @Before
    public void init() {

        //创建hadoop配置文件对象
        Configuration configuration = new Configuration();

        //指定zk地址
        configuration.set("hbase.zookeeper.quorum", "node1:2181,node2:2181,node3:2181");


        try {
            //建立连接   和zk的连接
            connection = HConnectionManager.createConnection(configuration);

            hBaseAdmin = new HBaseAdmin(configuration);

            System.out.println("连接建立成功");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 在由索引表的情况下查询数据
     */

    @Test
    public void queryIndex() {

        String clazz = "理科二班";

        /**
         * 1、先查询索引表
         * 2、再查询数据表
         *
         */


        try {
            HTableInterface student_index = connection.getTable("student_index");

            Get get = new Get(clazz.getBytes());
            Result result = student_index.get(get);

            ArrayList<Get> gets = new ArrayList<>();

            for (Cell cell : result.listCells()) {

                //取出列名  列名就是学号
                String id = Bytes.toString(CellUtil.cloneQualifier(cell));

                gets.add(new Get(id.getBytes()));

            }

            //查询学生表

            HTableInterface student = connection.getTable("student");

            //取出所有学生
            Result[] results = student.get(gets);


            for (Result result1 : results) {
                //取出rowkey
                String id = Bytes.toString(result1.getRow());


                // 1、直接通过列名取值   如果列名不存在 返回对应类型的默认值
                byte[] bytes = result1.getValue("info".getBytes(), "name".getBytes());
                //通过String的构造函数 构建字符串
                //String name = new String(bytes, "utf-8");

                //hbase  推荐方式
                String name = Bytes.toString(bytes);

                //int  类型  Bytes.toInt  转换
                int age = Bytes.toInt(result1.getValue("info".getBytes(), "age".getBytes()));

                String gender = Bytes.toString(result1.getValue("info".getBytes(), "gender".getBytes()));
                String clazz1 = Bytes.toString(result1.getValue("info".getBytes(), "clazz".getBytes()));

                System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz1);

            }


        } catch (IOException e) {
            e.printStackTrace();
        }

    }


    @Test
    public void put() {
        /**
         * 有了索引表之后插入数
         *
         * 1、插入索引表
         * 2、插入数据表
         *
         *
         * 1600100009,小伟,30,男,理科二班
         *
         */


        try {
            //保存索引
            HTableInterface student_index = connection.getTable("student_index");

            Put put = new Put("理科二班".getBytes());
            put.add("i".getBytes(), "1600100009".getBytes(), "".getBytes());

            student_index.put(put);


            //插入数据
            HTableInterface student = connection.getTable("student");

            Put put1 = new Put("1600100009".getBytes());
            put1.add("info".getBytes(), "name".getBytes(), "小伟".getBytes());
            put1.add("info".getBytes(), "gender".getBytes(), "男".getBytes());
            put1.add("info".getBytes(), "clazz".getBytes(), "理科二班".getBytes());
            put1.add("info".getBytes(), "age".getBytes(), Bytes.toBytes(30));


            student.put(put1);


        } catch (IOException e) {
            e.printStackTrace();
        }


    }

    @Test
    public void delete() {
        /**
         * 删除数据
         *
         * 1、删除索引表
         * 2、删除数据表
         *
         * 1600100009,小伟,30,男,理科二班
         *
         */


        try {
            //删除索引
            HTableInterface student_index = connection.getTable("student_index");

            Delete delete = new Delete("理科二班".getBytes());
            delete.deleteColumn("i".getBytes(), "1600100009".getBytes());

            student_index.delete(delete);

            //删除数据
            HTableInterface student = connection.getTable("student");

            Delete delete1 = new Delete("1600100009".getBytes());

            student.delete(delete1);


        } catch (IOException e) {
            e.printStackTrace();
        }


    }


    @Test
    public void put1() {
        /**
         * 有了协处理器之后put数据   索引会在服务端自动增加
         *
         *
         * 1600100009,小伟,30,男,理科二班
         *
         */


        try {
            //插入数据
            HTableInterface student = connection.getTable("student");

            Put put1 = new Put("1600100009".getBytes());
            put1.add("info".getBytes(), "name".getBytes(), "小伟".getBytes());
            put1.add("info".getBytes(), "gender".getBytes(), "男".getBytes());
            put1.add("info".getBytes(), "clazz".getBytes(), "理科二班".getBytes());
            put1.add("info".getBytes(), "age".getBytes(), Bytes.toBytes(30));


            student.put(put1);


        } catch (IOException e) {
            e.printStackTrace();
        }


    }


}
